# -*- coding: utf-8 -*-
"""
报表服务 - 核心业务逻辑
负责报表的编排、生成、文件管理等
"""
import os
import hashlib
import json
import logging
import shutil
import pymysql.cursors
from datetime import datetime
from typing import Dict, List, Optional, Any
from pathlib import Path

from app.utils.mysql_db import get_db_connection, is_database_connection_error
from app.services.data_source_service import DataSourceService


class ReportOrchestrator:
    """报表编排器 - 协调数据提供者和渲染器生成报表"""
    
    def __init__(self, reports_dir: str = None):
        """
        初始化报表编排器
        :param reports_dir: 报表文件存储目录
        """
        if reports_dir is None:
            # 默认存储在项目根目录的 reports 目录（便于前后端统一访问）
            backend_dir = Path(__file__).parent.parent.parent  # backend/
            project_root = backend_dir.parent                 # 项目根目录
            reports_dir = project_root / 'reports'
        
        self.reports_dir = Path(reports_dir)
        self.reports_dir.mkdir(parents=True, exist_ok=True)
    
    def create_run(self, template_code: str, params: Dict[str, Any], 
                   config_id: Optional[int] = None, 
                   trigger_type: str = 'manual',
                   title: Optional[str] = None,
                   output_formats: Optional[List[str]] = None) -> int:
        """
        创建报表运行记录
        :param template_code: 模板编码
        :param params: 运行参数
        :param config_id: 配置ID（可选）
        :param trigger_type: 触发类型（manual/scheduled/api）
        :param title: 报表标题（可选）
        :param output_formats: 输出格式列表（可选）
        :return: 运行ID
        """
        conn = get_db_connection()
        cursor = conn.cursor()
        
        try:
            def _normalize_time_value(value):
                if value is None:
                    return None
                if isinstance(value, str):
                    value = value.strip()
                    if value == '':
                        return None
                return value
            
            # 提取时间范围
            # 优先从 params 根级别获取
            # 注意：使用 get 方法，如果 key 不存在返回 None
            # 如果 key 存在但值为 None，也表示前端明确设置了"全部数据"
            time_start = _normalize_time_value(params.get('time_start') if 'time_start' in params else None)
            time_end = _normalize_time_value(params.get('time_end') if 'time_end' in params else None)
            
            # 如果根级别没有设置或值为空，尝试从嵌套配置中提取
            # 这样可以兼容旧的配置或直接从嵌套配置中传递的情况
            if time_start is None and time_end is None:
                # 对于 CPS 数据质量监控报表，从 time_filter 中提取
                if isinstance(params.get('time_filter'), dict):
                    time_filter = params.get('time_filter')
                    time_range_type = time_filter.get('time_range_type', 'all')
                    if time_range_type == 'all':
                        time_start = None
                        time_end = None
                    elif time_range_type == 'recent':
                        # 最近7天，计算开始和结束时间
                        from datetime import datetime, timedelta
                        time_end = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                        time_start = (datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d %H:%M:%S')
                    elif time_range_type in ('custom', 'yesterday', 'last_month'):
                        # custom、yesterday、last_month 都从 time_filter 中获取 time_start 和 time_end
                        time_start = time_filter.get('time_start')
                        time_end = time_filter.get('time_end')
                
                # 尝试从 source_tables 中提取
                if time_start is None and time_end is None and isinstance(params.get('source_tables'), dict):
                    source_tables = params.get('source_tables')
                    # 如果 time_range_type 为 "all"，time_start 和 time_end 应该为 None
                    if source_tables.get('time_range_type') == 'all':
                        time_start = None
                        time_end = None
                    else:
                        time_start = source_tables.get('time_start')
                        time_end = source_tables.get('time_end')
                
                # 如果还没有，尝试从 target_table_config 或类似配置中提取
                if time_start is None and time_end is None:
                    for key, value in params.items():
                        if isinstance(value, dict) and value.get('table_name'):
                            # 这可能是 target_table_config
                            # 如果 time_range_type 为 "all"，time_start 和 time_end 应该为 None
                            if value.get('time_range_type') == 'all':
                                time_start = None
                                time_end = None
                                break
                            else:
                                time_start = value.get('time_start')
                                time_end = value.get('time_end')
                                if time_start is not None and time_end is not None:
                                    break
            
            # 标准化参数格式，确保存储的参数格式一致
            normalized_params = self._normalize_params(params)
            
            # 如果title为空且config_id存在，从report_configs表查询config_name作为title
            if not title and config_id:
                try:
                    cursor.execute("SELECT config_name FROM report_configs WHERE id = %s", (config_id,))
                    result = cursor.fetchone()
                    if result and result[0]:
                        title = result[0]
                except Exception as e:
                    # 如果查询失败，使用None作为title
                    logger = logging.getLogger('flink.error')
                    logger.warning(f"查询配置名称失败: {e}, config_id: {config_id}")
            
            # 如果 time_start 或 time_end 是空字符串，转换为 None（表示全部数据）
            if time_start == '':
                time_start = None
            if time_end == '':
                time_end = None
            
            # 确保 trigger_type 不为空且有效（防止意外覆盖）
            if not trigger_type or trigger_type not in ['manual', 'scheduled', 'api']:
                trigger_type = 'manual'
                logger = logging.getLogger('flink.error')
                logger.warning(f"trigger_type 无效，使用默认值 'manual', config_id: {config_id}")
            
            # 插入运行记录
            sql = """
                INSERT INTO report_runs 
                (report_config_id, template_code, title, params, output_formats, time_start, time_end, 
                 trigger_type, status, progress)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            """
            cursor.execute(sql, (
                config_id, template_code, title, json.dumps(normalized_params, ensure_ascii=False),
                json.dumps(output_formats or ['html'], ensure_ascii=False),
                time_start, time_end, trigger_type, 'queued', 0
            ))
            conn.commit()
            
            # 记录日志，确保 trigger_type 正确
            logger = logging.getLogger('flink.business')
            logger.info(f"创建报表运行记录: run_id={cursor.lastrowid}, config_id={config_id}, trigger_type={trigger_type}")
            
            return cursor.lastrowid
        finally:
            cursor.close()
            conn.close()
    
    def _normalize_params(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """
        标准化参数格式，确保存储的参数格式一致
        :param params: 原始参数
        :return: 标准化后的参数
        """
        normalized = params.copy()
        
        # 将 platform_selection 转换为 platform_filter 以保持格式一致
        if 'platform_selection' in normalized:
            platform_selection = normalized['platform_selection']
            if isinstance(platform_selection, dict):
                normalized['platform_filter'] = platform_selection
                # 可选：保留原始的 platform_selection 以便调试
                # normalized.pop('platform_selection')
        
        return normalized
    
    def update_run_status(self, run_id: int, status: str, progress: int = None,
                         error_message: str = None, summary: Dict = None):
        """
        更新运行状态
        :param run_id: 运行ID
        :param status: 状态（queued/running/success/failed）
        :param progress: 进度（0-100）
        :param error_message: 错误信息
        :param summary: 概要信息
        """
        conn = get_db_connection()
        cursor = conn.cursor()
        
        try:
            updates = ['status = %s']
            values = [status]
            
            if progress is not None:
                updates.append('progress = %s')
                values.append(progress)
            
            if error_message is not None:
                updates.append('error_message = %s')
                values.append(error_message)
            
            if summary is not None:
                updates.append('summary = %s')
                values.append(json.dumps(summary, ensure_ascii=False))
            
            if status in ['success', 'failed']:
                updates.append('finished_at = NOW()')
            
            values.append(run_id)
            
            sql = f"UPDATE report_runs SET {', '.join(updates)} WHERE id = %s"
            cursor.execute(sql, values)
            conn.commit()
        finally:
            cursor.close()
            conn.close()
    
    def save_file(self, run_id: int, file_type: str, content: bytes,
                  previewable: bool = False) -> Dict[str, Any]:
        """
        保存报表文件
        :param run_id: 运行ID
        :param file_type: 文件类型（html/excel/csv/pdf）
        :param content: 文件内容（字节）
        :param previewable: 是否可预览
        :return: 文件信息
        """
        # 创建运行专属目录
        run_dir = self.reports_dir / str(run_id)
        run_dir.mkdir(parents=True, exist_ok=True)
        
        # 生成文件名
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        ext_map = {'html': 'html', 'excel': 'xlsx', 'csv': 'csv', 'pdf': 'pdf'}
        ext = ext_map.get(file_type, 'bin')
        filename = f"report_{timestamp}.{ext}"
        file_path = run_dir / filename
        
        # 写入文件
        with open(file_path, 'wb') as f:
            f.write(content)
        
        # 计算校验和
        checksum = hashlib.sha256(content).hexdigest()
        file_size = len(content)
        
        # 记录到数据库
        conn = get_db_connection()
        cursor = conn.cursor()
        
        try:
            sql = """
                INSERT INTO report_files 
                (run_id, file_type, file_path, file_size, checksum, previewable)
                VALUES (%s, %s, %s, %s, %s, %s)
            """
            # 存储相对路径
            relative_path = f"{run_id}/{filename}"
            cursor.execute(sql, (
                run_id, file_type, relative_path, file_size, checksum, previewable
            ))
            conn.commit()
            
            file_id = cursor.lastrowid
            
            return {
                'id': file_id,
                'file_type': file_type,
                'file_path': relative_path,
                'file_size': file_size,
                'checksum': checksum,
                'previewable': previewable
            }
        finally:
            cursor.close()
            conn.close()
    
    def get_file_path(self, file_id: int) -> Optional[Path]:
        """
        获取文件的绝对路径
        :param file_id: 文件ID
        :return: 文件路径
        """
        conn = get_db_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        try:
            sql = "SELECT file_path FROM report_files WHERE id = %s"
            cursor.execute(sql, (file_id,))
            result = cursor.fetchone()
            
            if result:
                return self.reports_dir / result['file_path']
            return None
        finally:
            cursor.close()
            conn.close()
    
    def generate_report(self, run_id: int, provider, renderer) -> bool:
        """
        生成报表（由具体的 provider 和 renderer 实现）
        注意：renderer 应该在调用此方法之前已经设置了正确的 title
        :param run_id: 运行ID
        :param provider: 数据提供者
        :param renderer: 渲染器（应已设置title）
        :return: 是否成功
        """
        try:
            self.update_run_status(run_id, 'running', progress=10)
            
            # 获取数据
            data = provider.collect()
            self.update_run_status(run_id, 'running', progress=50)
            
            # 渲染报表（renderer 中已包含 title）
            content = renderer.render(data)
            self.update_run_status(run_id, 'running', progress=80)
            
            # 保存文件
            file_type = renderer.get_file_type()
            previewable = file_type == 'html'
            self.save_file(run_id, file_type, content, previewable)
            
            # 生成摘要
            summary = provider.generate_summary(data)
            self.update_run_status(run_id, 'success', progress=100, summary=summary)
            
            return True
        except Exception as e:
            self.update_run_status(run_id, 'failed', progress=0, 
                                 error_message=str(e))
            return False


class ReportTemplateService:
    """报表模板服务"""
    
    @staticmethod
    def get_all_templates(group: str = None, is_active: bool = None,
                         page: int = None, page_size: int = None,
                         sort_field: str = 'template_group', 
                         sort_order: str = 'asc') -> List[Dict]:
        """
        获取所有模板
        :param group: 分组过滤
        :param is_active: 是否激活过滤
        :param page: 页码（可选，用于分页）
        :param page_size: 每页数量（可选，用于分页）
        :param sort_field: 排序字段
        :param sort_order: 排序方向 (asc/desc)
        :return: 模板列表
        """
        conn = get_db_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        try:
            sql = "SELECT * FROM report_templates WHERE 1=1"
            params = []
            
            if group:
                sql += " AND template_group = %s"
                params.append(group)
            
            if is_active is not None:
                sql += " AND is_active = %s"
                params.append(1 if is_active else 0)
            
            # 验证排序字段，防止SQL注入
            allowed_sort_fields = ['id', 'template_code', 'template_name', 'template_group', 
                                  'created_at', 'updated_at', 'version']
            if sort_field not in allowed_sort_fields:
                sort_field = 'template_group'
            
            sort_order = 'DESC' if sort_order.lower() == 'desc' else 'ASC'
            sql += f" ORDER BY {sort_field} {sort_order}, template_code"
            
            # 分页
            if page is not None and page_size is not None:
                offset = (page - 1) * page_size
                sql += " LIMIT %s OFFSET %s"
                params.extend([page_size, offset])
            
            try:
                cursor.execute(sql, params)
                rows = cursor.fetchall()
            except Exception as e:
                # 记录SQL与参数，便于排查生产问题
                import logging
                logger = logging.getLogger('flink.error')
                try:
                    logger.error(f"查询report_runs失败: {e} | SQL: {sql} | Params: {params}")
                except Exception:
                    pass
                raise
            
            # 格式化时间字段（数据库存储的是北京时间，直接格式化为字符串）
            for row in rows:
                if row.get('created_at') and hasattr(row['created_at'], 'strftime'):
                    row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S')
                if row.get('updated_at') and hasattr(row['updated_at'], 'strftime'):
                    row['updated_at'] = row['updated_at'].strftime('%Y-%m-%d %H:%M:%S')
            
            return rows
        finally:
            cursor.close()
            conn.close()
    
    @staticmethod
    def get_template_by_code(template_code: str) -> Optional[Dict]:
        """
        根据编码获取模板
        :param template_code: 模板编码
        :return: 模板信息
        """
        conn = get_db_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        try:
            sql = "SELECT * FROM report_templates WHERE template_code = %s"
            cursor.execute(sql, (template_code,))
            row = cursor.fetchone()
            
            # 格式化时间字段（数据库存储的是北京时间，直接格式化为字符串）
            if row:
                if row.get('created_at') and hasattr(row['created_at'], 'strftime'):
                    row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S')
                if row.get('updated_at') and hasattr(row['updated_at'], 'strftime'):
                    row['updated_at'] = row['updated_at'].strftime('%Y-%m-%d %H:%M:%S')
            
            return row
        finally:
            cursor.close()
            conn.close()
    
    @staticmethod
    def get_template_by_id(template_id: int) -> Optional[Dict]:
        """
        根据ID获取模板
        :param template_id: 模板ID
        :return: 模板信息
        """
        conn = get_db_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        try:
            sql = "SELECT * FROM report_templates WHERE id = %s"
            cursor.execute(sql, (template_id,))
            row = cursor.fetchone()
            
            # 格式化时间字段（数据库存储的是北京时间，直接格式化为字符串）
            if row:
                if row.get('created_at') and hasattr(row['created_at'], 'strftime'):
                    row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S')
                if row.get('updated_at') and hasattr(row['updated_at'], 'strftime'):
                    row['updated_at'] = row['updated_at'].strftime('%Y-%m-%d %H:%M:%S')
            
            return row
        finally:
            cursor.close()
            conn.close()
    
    @staticmethod
    def update_template(template_id: int, **kwargs) -> bool:
        """
        更新模板（仅允许更新有限字段）
        :param template_id: 模板ID
        :param kwargs: 可更新字段（template_name, description, is_active）
        :return: 是否成功
        """
        conn = get_db_connection()
        cursor = conn.cursor()
        
        try:
            # 白名单：仅允许更新这些字段
            allowed_fields = ['template_name', 'description', 'is_active']
            updates = []
            values = []
            
            for field in allowed_fields:
                if field in kwargs:
                    updates.append(f"{field} = %s")
                    values.append(kwargs[field])
            
            if not updates:
                return True  # 没有需要更新的字段
            
            updates.append('updated_at = NOW()')
            values.append(template_id)
            
            sql = f"UPDATE report_templates SET {', '.join(updates)} WHERE id = %s"
            cursor.execute(sql, values)
            conn.commit()
            
            return cursor.rowcount > 0
        finally:
            cursor.close()
            conn.close()
    
    @staticmethod
    def create_template(template_code: str, template_name: str, 
                       template_group: str, param_schema: Dict,
                       default_params: Dict, output_formats: List[str],
                       description: str = None) -> int:
        """
        创建模板
        :return: 模板ID
        """
        conn = get_db_connection()
        cursor = conn.cursor()
        
        try:
            sql = """
                INSERT INTO report_templates 
                (template_code, template_name, template_group, description,
                 param_schema, default_params, output_formats)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
            """
            cursor.execute(sql, (
                template_code, template_name, template_group, description,
                json.dumps(param_schema, ensure_ascii=False),
                json.dumps(default_params, ensure_ascii=False),
                json.dumps(output_formats, ensure_ascii=False)
            ))
            conn.commit()
            return cursor.lastrowid
        finally:
            cursor.close()
            conn.close()


class ReportConfigService:
    """报表配置服务"""
    
    @staticmethod
    def get_all_configs(template_code: str = None, is_active: bool = None,
                       schedule_enabled: bool = None,
                       sort_field: str = 'created_at', sort_order: str = 'desc') -> List[Dict]:
        """
        获取所有配置
        :param template_code: 模板编码过滤
        :param is_active: 是否激活过滤
        :param schedule_enabled: 定时任务是否启用过滤
        :param sort_field: 排序字段
        :param sort_order: 排序方向 (asc/desc)
        :return: 配置列表
        """
        conn = None
        cursor = None
        
        try:
            conn = get_db_connection()
            cursor = conn.cursor(pymysql.cursors.DictCursor)
            
            sql = "SELECT * FROM report_configs WHERE 1=1"
            params = []
            
            if template_code:
                sql += " AND template_code = %s"
                params.append(template_code)
            
            if is_active is not None:
                sql += " AND is_active = %s"
                params.append(1 if is_active else 0)
            
            if schedule_enabled is not None:
                sql += " AND schedule_enabled = %s"
                params.append(1 if schedule_enabled else 0)
            
            # 验证排序字段，防止SQL注入
            allowed_sort_fields = ['id', 'config_name', 'template_code', 'created_at', 'updated_at', 
                                  'is_active', 'schedule_enabled']
            if sort_field not in allowed_sort_fields:
                sort_field = 'created_at'
            
            # 验证排序方向
            sort_order = 'DESC' if sort_order.lower() == 'desc' else 'ASC'
            
            sql += f" ORDER BY {sort_field} {sort_order}"
            
            cursor.execute(sql, params)
            rows = cursor.fetchall()
            
            # 格式化时间字段和解析JSON字段
            for row in rows:
                if row.get('created_at') and hasattr(row['created_at'], 'strftime'):
                    row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S')
                if row.get('updated_at') and hasattr(row['updated_at'], 'strftime'):
                    row['updated_at'] = row['updated_at'].strftime('%Y-%m-%d %H:%M:%S')
                
                # 解析params JSON，支持data_source_ids
                if row.get('params'):
                    try:
                        params_data = json.loads(row['params']) if isinstance(row['params'], str) else row['params']
                        row['params'] = ReportConfigService.parse_params_with_data_sources(params_data)
                    except Exception as e:
                        # JSON解析失败，保持原值
                        logger = logging.getLogger('flink.error')
                        logger.warning(f"解析配置params JSON失败: {e}, 配置ID: {row.get('id')}")
                
                # 解析output_formats JSON
                if row.get('output_formats'):
                    try:
                        row['output_formats'] = json.loads(row['output_formats']) if isinstance(row['output_formats'], str) else row['output_formats']
                    except Exception as e:
                        # JSON解析失败，保持原值
                        logger = logging.getLogger('flink.error')
                        logger.warning(f"解析配置output_formats JSON失败: {e}, 配置ID: {row.get('id')}")
            
            return rows
            
        except Exception as e:
            # 重新抛出异常，让调用方处理
            # 调用方（API端点）会使用 is_database_connection_error 来处理错误
            raise
        finally:
            if cursor:
                try:
                    cursor.close()
                except:
                    pass
            if conn:
                try:
                    conn.close()
                except:
                    pass
    
    @staticmethod
    def get_config_by_id(config_id: int) -> Optional[Dict]:
        """
        根据ID获取配置
        :param config_id: 配置ID
        :return: 配置信息
        """
        conn = None
        cursor = None
        
        try:
            conn = get_db_connection()
            cursor = conn.cursor(pymysql.cursors.DictCursor)
            
            sql = "SELECT * FROM report_configs WHERE id = %s"
            cursor.execute(sql, (config_id,))
            row = cursor.fetchone()
            
            # 格式化时间字段和解析JSON字段
            if row:
                if row.get('created_at') and hasattr(row['created_at'], 'strftime'):
                    row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S')
                if row.get('updated_at') and hasattr(row['updated_at'], 'strftime'):
                    row['updated_at'] = row['updated_at'].strftime('%Y-%m-%d %H:%M:%S')
                
                # 解析params JSON，支持data_source_ids
                if row.get('params'):
                    try:
                        params_data = json.loads(row['params']) if isinstance(row['params'], str) else row['params']
                        row['params'] = ReportConfigService.parse_params_with_data_sources(params_data)
                    except Exception as e:
                        # JSON解析失败，保持原值
                        logger = logging.getLogger('flink.error')
                        logger.warning(f"解析配置params JSON失败: {e}, 配置ID: {config_id}")
                
                # 解析output_formats JSON
                if row.get('output_formats'):
                    try:
                        row['output_formats'] = json.loads(row['output_formats']) if isinstance(row['output_formats'], str) else row['output_formats']
                    except Exception as e:
                        # JSON解析失败，保持原值
                        logger = logging.getLogger('flink.error')
                        logger.warning(f"解析配置output_formats JSON失败: {e}, 配置ID: {config_id}")
            
            return row
            
        except Exception as e:
            # 重新抛出异常，让调用方处理
            # 调用方（API端点）会使用 is_database_connection_error 来处理错误
            raise
        finally:
            if cursor:
                try:
                    cursor.close()
                except:
                    pass
            if conn:
                try:
                    conn.close()
                except:
                    pass
    
    @staticmethod
    def create_config(config_name: str, template_code: str, params: Dict,
                     output_formats: List[str], schedule_enabled: bool = False,
                     schedule_cron: str = None, notify_on_success: bool = False) -> int:
        """
        创建配置
        :return: 配置ID
        """
        # 验证和解析数据源配置
        ReportConfigService._validate_data_source_ids(params)
        
        # 标准化参数格式
        normalized_params = ReportConfigService._normalize_params(params)
        
        conn = get_db_connection()
        cursor = conn.cursor()
        
        try:
            sql = """
                INSERT INTO report_configs 
                (config_name, template_code, params, output_formats,
                 schedule_enabled, schedule_cron, notify_on_success)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
            """
            cursor.execute(sql, (
                config_name, template_code,
                json.dumps(normalized_params, ensure_ascii=False),
                json.dumps(output_formats, ensure_ascii=False),
                1 if schedule_enabled else 0,
                schedule_cron,
                1 if notify_on_success else 0
            ))
            conn.commit()
            return cursor.lastrowid
        finally:
            cursor.close()
            conn.close()
    
    @staticmethod
    def update_config(config_id: int, **kwargs) -> bool:
        """
        更新配置
        :param config_id: 配置ID
        :param kwargs: 要更新的字段
        :return: 是否成功
        """
        if not kwargs:
            return False
        
        # 如果更新params，验证数据源配置
        if 'params' in kwargs:
            ReportConfigService._validate_data_source_ids(kwargs['params'])
            # 标准化参数格式
            kwargs['params'] = ReportConfigService._normalize_params(kwargs['params'])
        
        conn = get_db_connection()
        cursor = conn.cursor()
        
        try:
            updates = []
            values = []
            
            for key, value in kwargs.items():
                if key in ['params', 'output_formats']:
                    updates.append(f"{key} = %s")
                    values.append(json.dumps(value, ensure_ascii=False))
                elif key in ['schedule_enabled', 'notify_on_success', 'is_active']:
                    updates.append(f"{key} = %s")
                    values.append(1 if value else 0)
                else:
                    updates.append(f"{key} = %s")
                    values.append(value)
            
            values.append(config_id)
            
            sql = f"UPDATE report_configs SET {', '.join(updates)} WHERE id = %s"
            cursor.execute(sql, values)
            conn.commit()
            return cursor.rowcount > 0
        finally:
            cursor.close()
            conn.close()
    
    @staticmethod
    def _validate_data_source_ids(params: Dict) -> None:
        """
        验证数据源ID列表
        :param params: 参数字典
        :raises ValueError: 如果数据源ID无效
        """
        if 'data_source_ids' not in params:
            return
        
        data_source_ids = params['data_source_ids']
        if not data_source_ids:
            return
        
        # 确保是列表
        if not isinstance(data_source_ids, list):
            raise ValueError("data_source_ids必须是列表类型")
        
        # 验证每个ID是否存在且为mysql或starrocks类型
        for config_id in data_source_ids:
            try:
                config = DataSourceService.get_config(config_id)
                if config['service_type'] not in ['mysql', 'starrocks']:
                    raise ValueError(
                        f"数据源配置 {config_id} 的类型为 {config['service_type']}，"
                        f"仅支持mysql和starrocks"
                    )
            except ValueError as e:
                raise ValueError(f"数据源配置 {config_id} 无效: {str(e)}")
    
    @staticmethod
    def _normalize_params(params: Dict[str, Any]) -> Dict[str, Any]:
        """
        标准化参数格式，确保存储的参数格式一致
        :param params: 原始参数
        :return: 标准化后的参数
        """
        normalized = params.copy()
        
        # 将 platform_selection 转换为 platform_filter 以保持格式一致
        if 'platform_selection' in normalized:
            platform_selection = normalized['platform_selection']
            if isinstance(platform_selection, dict):
                normalized['platform_filter'] = platform_selection
                # 可选：保留原始的 platform_selection 以便调试
                # normalized.pop('platform_selection')
        
        return normalized
    
    @staticmethod
    def parse_params_with_data_sources(params: Dict) -> Dict:
        """
        解析参数，将data_source_ids转换为数据源配置信息（用于前端展示）
        :param params: 参数字典
        :return: 解析后的参数字典（包含data_source_configs字段）
        """
        result = params.copy()
        
        if 'data_source_ids' in params and params.get('data_source_ids'):
            try:
                data_source_configs = DataSourceService.get_configs(
                    config_ids=params['data_source_ids']
                )
                result['data_source_configs'] = data_source_configs
            except Exception as e:
                # 如果获取失败，记录错误但不影响其他参数
                result['data_source_configs'] = []
                result['_data_source_error'] = str(e)
        
        return result
    
    @staticmethod
    def delete_config(config_id: int) -> bool:
        """
        删除配置
        :param config_id: 配置ID
        :return: 是否成功
        """
        conn = get_db_connection()
        cursor = conn.cursor()
        
        try:
            sql = "DELETE FROM report_configs WHERE id = %s"
            cursor.execute(sql, (config_id,))
            conn.commit()
            return cursor.rowcount > 0
        finally:
            cursor.close()
            conn.close()


class ReportRunService:
    """报表运行服务"""
    
    @staticmethod
    def update_run_notification(run_id: int, notify_requested: bool = None,
                                notify_channels: Optional[List[str]] = None,
                                notify_params: Optional[Dict[str, Any]] = None) -> None:
        """
        更新运行记录的通知元数据
        :param run_id: 运行ID
        :param notify_requested: 是否请求发送通知
        :param notify_channels: 使用的通知渠道列表
        :param notify_params: 通知参数（解析后的JSON结构）
        """
        updates = []
        values = []

        if notify_requested is not None:
            updates.append('notify_requested = %s')
            values.append(1 if notify_requested else 0)

        if notify_channels is not None:
            updates.append('notify_channels = %s')
            values.append(json.dumps(notify_channels, ensure_ascii=False))

        if notify_params is not None:
            updates.append('notify_params = %s')
            values.append(json.dumps(notify_params, ensure_ascii=False))

        if not updates:
            return

        conn = get_db_connection()
        cursor = conn.cursor()

        try:
            values.append(run_id)
            sql = f"UPDATE report_runs SET {', '.join(updates)} WHERE id = %s"
            cursor.execute(sql, values)
            conn.commit()
        finally:
            cursor.close()
            conn.close()

    @staticmethod
    def get_runs(template_code: str = None, status: str = None,
                trigger_type: str = None, start_date: str = None, end_date: str = None,
                limit: int = 100, offset: int = 0,
                sort_field: str = 'created_at', sort_order: str = 'desc',
                report_config_id: int = None, include_all: bool = False,
                run_id: int = None) -> List[Dict]:
        """
        获取运行历史
        :param template_code: 模板编码过滤
        :param status: 状态过滤
        :param trigger_type: 触发方式过滤
        :param start_date: 开始时间过滤
        :param end_date: 结束时间过滤
        :param limit: 限制数量
        :param offset: 偏移量
        :param sort_field: 排序字段
        :param sort_order: 排序方向 (asc/desc)
        :param report_config_id: 报表配置ID过滤
        :param include_all: 是否包含一次性运行（需要配合template_code使用）
        :param run_id: 运行ID过滤（精确匹配单条记录，优先级最高）
        :return: 运行历史列表
        """
        conn = get_db_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        try:
            # 构建条件
            base_conditions = []
            params = []
            
            # run_id 优先级最高，如果提供了 run_id，直接根据 id 筛选，不再应用其他筛选条件
            if run_id:
                base_conditions.append("id = %s")
                params.append(run_id)
                # 当有 run_id 时，不再应用其他筛选条件（status、trigger_type 等）
                # 因为 run_id 是精确匹配单条记录
            else:
                # 只有当没有 run_id 时，才应用其他筛选条件
                if report_config_id:
                    if include_all and template_code:
                        # 该任务下的所有运行 + 同模板的一次性运行
                        base_conditions.append("(report_config_id = %s OR (template_code = %s AND report_config_id IS NULL))")
                        params.extend([report_config_id, template_code])
                    else:
                        # 只根据 report_config_id 筛选
                        base_conditions.append("report_config_id = %s")
                        params.append(report_config_id)
                
                if template_code and not report_config_id:
                    base_conditions.append("template_code = %s")
                    params.append(template_code)
                
                # 只有当没有 run_id 时，才应用 status、trigger_type 等筛选条件
                if status:
                    base_conditions.append("status = %s")
                    params.append(status)
                
                if trigger_type:
                    base_conditions.append("trigger_type = %s")
                    params.append(trigger_type)
                
                if start_date:
                    base_conditions.append("created_at >= %s")
                    params.append(start_date)
                
                if end_date:
                    base_conditions.append("created_at <= %s")
                    params.append(end_date)
            
            where_clause = " AND ".join(base_conditions) if base_conditions else "1=1"
            
            # 验证排序字段，防止SQL注入
            allowed_sort_fields = ['id', 'template_code', 'created_at', 'finished_at', 'status', 'trigger_type', 'report_config_id']
            if sort_field not in allowed_sort_fields:
                sort_field = 'created_at'
            
            # 验证排序方向
            sort_order = 'DESC' if sort_order.lower() == 'desc' else 'ASC'
            
            sql = f"SELECT * FROM report_runs WHERE {where_clause} ORDER BY {sort_field} {sort_order} LIMIT %s OFFSET %s"
            params.extend([limit, offset])
            
            cursor.execute(sql, params)
            rows = cursor.fetchall()
            
            # 格式化时间字段（数据库存储的是北京时间，直接格式化为字符串）
            for row in rows:
                if row.get('created_at') and hasattr(row['created_at'], 'strftime'):
                    row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S')
                if row.get('finished_at') and hasattr(row['finished_at'], 'strftime'):
                    row['finished_at'] = row['finished_at'].strftime('%Y-%m-%d %H:%M:%S')
                # 处理 time_start 和 time_end：如果是 None 或空字符串，保持为 None（前端会显示为"全部数据"）
                if row.get('time_start') and hasattr(row['time_start'], 'strftime'):
                    row['time_start'] = row['time_start'].strftime('%Y-%m-%d %H:%M:%S')
                elif row.get('time_start') is None or row.get('time_start') == '':
                    row['time_start'] = None
                if row.get('time_end') and hasattr(row['time_end'], 'strftime'):
                    row['time_end'] = row['time_end'].strftime('%Y-%m-%d %H:%M:%S')
                elif row.get('time_end') is None or row.get('time_end') == '':
                    row['time_end'] = None
                # 确保 title 字段不为空字符串时返回 None（便于前端判断）
                if row.get('title') == '':
                    row['title'] = None

                if row.get('params'):
                    try:
                        row['params'] = json.loads(row['params']) if isinstance(row['params'], str) else row['params']
                    except Exception:
                        pass

                if row.get('notify_channels'):
                    try:
                        row['notify_channels'] = json.loads(row['notify_channels']) if isinstance(row['notify_channels'], str) else row['notify_channels']
                    except Exception:
                        pass

                if row.get('notify_params'):
                    try:
                        row['notify_params'] = json.loads(row['notify_params']) if isinstance(row['notify_params'], str) else row['notify_params']
                    except Exception:
                        pass
            
            return rows
        finally:
            cursor.close()
            conn.close()
    
    @staticmethod
    def get_run_by_id(run_id: int) -> Optional[Dict]:
        """
        根据ID获取运行记录
        :param run_id: 运行ID
        :return: 运行记录
        """
        conn = get_db_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        try:
            sql = "SELECT * FROM report_runs WHERE id = %s"
            cursor.execute(sql, (run_id,))
            row = cursor.fetchone()
            
            # 格式化时间字段（数据库存储的是北京时间，直接格式化为字符串）
            if row:
                if row.get('created_at') and hasattr(row['created_at'], 'strftime'):
                    row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S')
                if row.get('finished_at') and hasattr(row['finished_at'], 'strftime'):
                    row['finished_at'] = row['finished_at'].strftime('%Y-%m-%d %H:%M:%S')
                # 处理 time_start 和 time_end：如果是 None 或空字符串，保持为 None（前端会显示为"全部数据"）
                if row.get('time_start') and hasattr(row['time_start'], 'strftime'):
                    row['time_start'] = row['time_start'].strftime('%Y-%m-%d %H:%M:%S')
                elif row.get('time_start') is None or row.get('time_start') == '':
                    row['time_start'] = None
                if row.get('time_end') and hasattr(row['time_end'], 'strftime'):
                    row['time_end'] = row['time_end'].strftime('%Y-%m-%d %H:%M:%S')
                elif row.get('time_end') is None or row.get('time_end') == '':
                    row['time_end'] = None
                # 确保 title 字段不为空字符串时返回 None（便于前端判断）
                if row.get('title') == '':
                    row['title'] = None
                
                # 从 params 中提取 data_source_id（如果存在）
                if row.get('params'):
                    try:
                        params = json.loads(row['params']) if isinstance(row['params'], str) else row['params']
                        # 尝试从 source_tables 或 target_tables 中提取 data_source_id
                        data_source_id = None
                        if isinstance(params, dict):
                            # 优先从 source_tables 中提取
                            if params.get('source_tables') and isinstance(params['source_tables'], dict):
                                data_source_id = params['source_tables'].get('data_source_id')
                            # 如果 source_tables 没有，尝试从 target_tables 中提取
                            if not data_source_id:
                                if params.get('target_tables') and isinstance(params['target_tables'], dict):
                                    data_source_id = params['target_tables'].get('data_source_id')
                            # 如果还没有，尝试遍历所有字段（支持动态字段名）
                            if not data_source_id:
                                for key, value in params.items():
                                    if isinstance(value, dict) and value.get('data_source_id'):
                                        data_source_id = value.get('data_source_id')
                                        break
                        
                        if data_source_id:
                            row['data_source_id'] = data_source_id
                    except Exception as e:
                        # 如果解析失败，忽略错误
                        logger = logging.getLogger(__name__)
                        logger.warning(f"解析 params 提取 data_source_id 失败: {str(e)}")
                    else:
                        row['params'] = params

                if row.get('notify_channels'):
                    try:
                        row['notify_channels'] = json.loads(row['notify_channels']) if isinstance(row['notify_channels'], str) else row['notify_channels']
                    except Exception:
                        pass

                if row.get('notify_params'):
                    try:
                        row['notify_params'] = json.loads(row['notify_params']) if isinstance(row['notify_params'], str) else row['notify_params']
                    except Exception:
                        pass
            
            return row
        finally:
            cursor.close()
            conn.close()
    
    @staticmethod
    def get_run_files(run_id: int) -> List[Dict]:
        """
        获取运行生成的文件列表
        :param run_id: 运行ID
        :return: 文件列表
        """
        conn = get_db_connection()
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        try:
            sql = "SELECT * FROM report_files WHERE run_id = %s ORDER BY created_at DESC"
            cursor.execute(sql, (run_id,))
            rows = cursor.fetchall()
            
            # 格式化时间字段
            for row in rows:
                if row.get('created_at') and hasattr(row['created_at'], 'strftime'):
                    row['created_at'] = row['created_at'].strftime('%Y-%m-%d %H:%M:%S')
            
            return rows
        finally:
            cursor.close()
            conn.close()
    
    @staticmethod
    def delete_run(run_id: int, remove_files: bool = True) -> bool:
        """
        删除运行记录及其关联文件
        :param run_id: 运行ID
        :param remove_files: 是否删除生成的物理文件
        :return: 是否删除成功
        """
        conn = None
        cursor = None
        file_records: List[Dict[str, Any]] = []
        try:
            conn = get_db_connection()
            cursor = conn.cursor(pymysql.cursors.DictCursor)
            
            # 查询文件记录，便于后续删除物理文件
            cursor.execute("SELECT id, file_path FROM report_files WHERE run_id = %s", (run_id,))
            file_records = cursor.fetchall()
            
            # 先删除文件记录，再删除运行记录
            cursor.execute("DELETE FROM report_files WHERE run_id = %s", (run_id,))
            cursor.execute("DELETE FROM report_runs WHERE id = %s", (run_id,))
            run_deleted = cursor.rowcount > 0
            conn.commit()
            
            if remove_files and run_deleted:
                orchestrator = ReportOrchestrator()
                logger = logging.getLogger('flink.business')
                
                # 删除数据库中记录到的单个文件
                for record in file_records:
                    relative_path = record.get('file_path')
                    if not relative_path:
                        continue
                    file_path = orchestrator.reports_dir / relative_path
                    try:
                        if file_path.exists():
                            file_path.unlink()
                            logger.info(f"删除报表文件成功: run_id={run_id}, file_path={file_path}")
                    except Exception as file_error:
                        logger.warning(f"删除报表文件失败: run_id={run_id}, file_path={file_path}, error={file_error}")
                
                # 删除运行目录（保证残留文件被清理）
                run_dir = orchestrator.reports_dir / str(run_id)
                if run_dir.exists():
                    try:
                        shutil.rmtree(run_dir)
                        logger.info(f"删除报表目录成功: run_id={run_id}, dir={run_dir}")
                    except Exception as dir_error:
                        logger.warning(f"删除报表目录失败: run_id={run_id}, dir={run_dir}, error={dir_error}")
            
            return run_deleted
        finally:
            if cursor:
                try:
                    cursor.close()
                except Exception:
                    pass
            if conn:
                try:
                    conn.close()
                except Exception:
                    pass

